package com.hp.bon.sgw.core.channel;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamReader;

import org.apache.commons.lang.StringEscapeUtils;
import org.dom4j.Document;
import org.dom4j.Element;
import org.simpleframework.http.ContentType;
import org.simpleframework.http.Request;
import org.simpleframework.http.Response;
import org.simpleframework.http.core.Container;
import org.simpleframework.http.core.ContainerProcessor;
import org.simpleframework.http.core.ContainerSelector;
import org.simpleframework.http.core.ContainerServer;
import org.simpleframework.transport.connect.Connection;
import org.simpleframework.transport.connect.SocketConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.hp.bon.sgw.bean.BlockRequestBean;
import com.hp.bon.sgw.bean.ProtocalConvertResult;
import com.hp.bon.sgw.bean.XmlMessage;
import com.hp.bon.sgw.convert.DataExchangerImpl;
import com.hp.bon.sgw.core.BuzProcessorManager;
import com.hp.bon.sgw.core.CallParam;
import com.hp.bon.sgw.core.InprogressMessageQueue;
import com.hp.bon.sgw.core.channel.handler.BaseWSCallHandler;
import com.hp.bon.sgw.core.channel.handler.DCCAsynCallHandler;
import com.hp.bon.sgw.core.channel.handler.TuxedoCallHandler;
import com.hp.bon.sgw.domain.MessageDetail;
import com.hp.bon.sgw.domain.Node;
import com.hp.bon.sgw.domain.SGWChannel;
import com.hp.bon.sgw.service.ConfigListener;
import com.hp.bon.sgw.service.ConfigService;
import com.hp.bon.sgw.service.RSASignService;
import com.hp.bon.sgw.start.SpringContext;
import com.hp.bon.sgw.stdbss30.OutApiBizProcedure;
import com.hp.bon.sgw.stdbss30.common.BizConsts;
import com.hp.bon.sgw.threads.BusinessRecordCacheManager;
import com.hp.bon.sgw.threads.MessageDetailCacheManager;
import com.hp.bon.sgw.util.Comutils;
import com.hp.bon.sgw.util.LogTool;
import com.hp.bon.sgw.util.RSAUtil;
import com.hp.bon.sgw.util.dcc.DccAvpHelper;
import com.hp.bon.sgw.wh.service.UserResourceConfigService;
import com.hp.ocs.diameter.message.AVP;
import com.hp.ocs.diameter.message.Message;

import cn.chinatelecom.sgw.soap.SGWService;
import cn.chinatelecom.sgw.soap.SgwQueryAck;
import cn.chinatelecom.sgw.soap.SgwQueryReq;

@javax.ws.rs.Path("/")
public class SoapChannel
  implements Container, ConfigListener, IChannel
{
  private static final Logger logger = LoggerFactory.getLogger(SoapChannel.class);
  static DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
  protected ConfigService confSrv;
  private int maxLinkCount;
  static JAXBContext jc = null;
  private boolean replaceRealm;
  private String standardWSReqPath = "/ws/SGWService";
  private String standardHttpReqPath = "/http/call";

  private ContainerServer server = null;
  protected RSASignService rsaSignService;
  static XMLInputFactory inputFactory = XMLInputFactory.newInstance();

  protected final Set<String> monitorConfKeys = new HashSet();
  private SGWChannel channel;
  private ConcurrentHashMap<String, DelayCallParam> _sessionMap = new ConcurrentHashMap();
  ThreadPoolExecutor executor;
  ThreadPoolExecutor collector;
  
  // 新帐务中心接口处理流程 类
  public static OutApiBizProcedure outApiBizProcedure;

  static
  {
    try
    {
      jc = JAXBContext.newInstance("cn.chinatelecom.sgw.soap");
    } catch (JAXBException e) {
      e.printStackTrace();
    }
  }

  public SoapChannel(SGWChannel channel)
  {
    this.channel = channel;
    this.confSrv = SpringContext.getInstance().getConfigService();
    this.confSrv.addConfigListener(this);
    Timer _timer = new Timer("DelayTask-" + channel.toString(), true);
    _timer.schedule(new DelayTask(), 1000L, 1000L);
  }

  public void start() {
    logger.info("start soap channel:" + this.channel.toString() + " ...");
    this.monitorConfKeys.add("business.max_conn");
    loadParams();
    String key = "sgw.webserver.thread.max";
    int maxThreads = this.confSrv.getIntValue(key, 50);
    logger.info(key + " :" + maxThreads);
    String endPoint = this.channel.getUri();
    if (endPoint == null) {
      logger.warn("channel dont't defined ws addr");
      return;
    }
    logger.info("start sgw webservice server portal at " + endPoint);
    
    // 初始化对外请求新帐务中心流程类，主要处理功能：对外计费3.0标准接口、对外查询中心接口、对外计费其他接口
    outApiBizProcedure = OutApiBizProcedure.instance();
    //UserResourceConfigService urcService = null;
    //urcService = SpringContext.getApplicationContext().getBean(UserResourceConfigService.class);
    //outApiBizProcedure.setUserResourceConfigService(urcService);
    logger.info("outApiBizProcedure init successfull.");
    
    try
    {
      URL srvURL = new URL(endPoint);
      int port = srvURL.getPort();
      if (port == -1) {
        port = 8080;
      }
      String host = srvURL.getHost();
      if ((host.equals("127.0.0.1")) || (host.equalsIgnoreCase("localhost"))) {
        host = "0.0.0.0";
      }
      this.server = new ContainerServer(this, maxThreads);

      Connection connection = new SocketConnection(this.server);
      SocketAddress address = new InetSocketAddress(host, port);
      connection.connect(address);
      this.collector = ((ContainerSelector)((ContainerProcessor)this.server.getProcessor()).getSelector()).getCollect().getQueue();
      this.executor = ((ContainerSelector)((ContainerProcessor)this.server.getProcessor()).getSelector()).getExecutor().getQueue();

      this.channel.setStatus(1);
      this.confSrv.updateChannelStatus2DB(this.channel.getId(), 1);
    } catch (IOException e) {
      logger.error("start soap channel " + this.channel.toString() + " err " + e);
    }
  }

  public static void directAck(XmlMessage ack, PrintStream responseStream)
    throws Exception
  {
    String orinData = ack.getDATA();
    String dataStr = "<![CDATA[" + orinData + "]]>";
    String errMsg = ack.getDS();
    errMsg = (errMsg != null) && (errMsg.indexOf('&') == -1) && (errMsg.indexOf('>') == -1) && (errMsg.indexOf('<') == -1) ? errMsg : StringEscapeUtils.escapeXml(errMsg);
    String respXML = "<SOAP-ENV:Envelope xmlns:SOAP-ENV=\"http://schemas.xmlsoap.org/soap/envelope/\"><SOAP-ENV:Header/><SOAP-ENV:Body><ns3:SgwQueryAck xmlns:ns3=\"http://www.chinatelecom.cn/SGW/SOAP\"><MID>" + ack.getMID() + "</MID><REPTAG>0</REPTAG><RTT>" + ack.getRTT() + "</RTT><SIG>" + 
      ack.getSIG() + "</SIG><RC>" + ack.getRC() + "</RC><DS>" + errMsg + "</DS><DATA>" + dataStr + "</DATA></ns3:SgwQueryAck></SOAP-ENV:Body></SOAP-ENV:Envelope>";
    responseStream.write(respXML.getBytes("UTF-8"));
    responseStream.close();
    if (logger.isInfoEnabled())
      logger.info("http response  " + respXML);
  }

  private void writeHTTPAck(String sig, String rc, String ds, String data, PrintStream out)
  {
    String ack = Comutils.genTuxedoRspsXmlStr(sig, rc, ds, data);
    logger.info("send http ack " + ack);
    try {
      out.write(ack.getBytes("UTF-8"));
      out.close();
    } catch (Exception e) {
      logger.warn("caught err:" + e);
    }
  }

  public static XmlMessage parseWSRequest(String representation) throws Exception
  {
    InputStream input = new ByteArrayInputStream(representation.getBytes("UTF-8"));
    XMLStreamReader xmlStreamReader = inputFactory.createXMLStreamReader(input);
    XmlMessage req = new XmlMessage(true);
    req.setLength(representation.getBytes().length);
    int nextType = 0;

    boolean seenSgwQueryReq = false;
    while (xmlStreamReader.hasNext()) {
      nextType = xmlStreamReader.next();
      if (nextType == 1) {
        String elName = xmlStreamReader.getLocalName();
        if (elName.equals("SgwQueryReq")) {
          seenSgwQueryReq = true;
        } else if (seenSgwQueryReq) {
          String value = xmlStreamReader.getElementText();

          if (elName.equals("MID"))
            req.setMID(value);
          else if (elName.equals("SCID"))
            req.setSCID(value);
          else if (elName.equals("UID"))
            req.setUID(value);
          else if (elName.equals("PWD"))
            req.setPWD(value);
          else if (elName.equals("SIG"))
            req.setSIG(value);
          else if (elName.equals("DATA")) {
            req.setDATA(value);
          }
        }
      }
    }

    return req;
  }

  public static XmlMessage parseWSResponse(String response) throws Exception {
    InputStream input = new ByteArrayInputStream(response.getBytes("UTF-8"));
    XMLStreamReader xmlStreamReader = inputFactory.createXMLStreamReader(input);
    XmlMessage ack = new XmlMessage(false);
    int nextType = 0;

    boolean seenSgwQueryAck = false;
    while (xmlStreamReader.hasNext()) {
      nextType = xmlStreamReader.next();
      if (nextType == 1) {
        String elName = xmlStreamReader.getLocalName();
        if (elName.equals("SgwQueryAck")) {
          seenSgwQueryAck = true;
        } else if (seenSgwQueryAck) {
          String value = xmlStreamReader.getElementText();
          if (elName.equals("MID"))
            ack.setMID(value);
          else if (elName.equals("RTT"))
            ack.setRTT(value);
          else if (elName.equals("RC"))
            ack.setRC(value);
          else if (elName.equals("SIG"))
            ack.setSIG(value);
          else if (elName.equals("DATA"))
            ack.setDATA(value);
          else if (elName.equals("DS")) {
            ack.setDS(value);
          }
        }
      }
    }
    return ack;
  }

  public static XmlMessage createERRAnwser(String msgID, int retCode, String errMsg) {
    XmlMessage ack = new XmlMessage(false);
    ack.setMID(msgID);
    ack.setRTT(new SimpleDateFormat("yyyyMMddHHmmssSSS").format(Long.valueOf(System.currentTimeMillis())));
    ack.setRC(Integer.toString(retCode));
    ack.setDS(errMsg);
    return ack;
  }

  public void handle(Request request, Response response)
  {
    long receiveTime = System.currentTimeMillis();
    PrintStream body = null;
    try {
      body = response.getPrintStream();
      response.set("Content-Type", "text/xml;charset=utf-8");
      response.set("Server", "HP  Asyn WS WebServer )");
      response.setDate("Date", receiveTime);
      response.setDate("Last-Modified", receiveTime);
      ContentType cntType = request.getContentType();
      if (cntType == null) {
        body.println("invalid request Content type is null");
        body.close();
        return;
      }
      cntType.setCharset(Comutils.fixContentCharSet(cntType.getCharset()));
      String reqContent = request.getContent();
      if (logger.isInfoEnabled()) {
        logger.info("http request from " + request.getClientAddress() + ", " + request.getPath().getPath() + "\r\n" + reqContent);
      }

      double persent = getSGWChannel().getRecvQueuePercent();
      double MIN = SpringContext.getInstance().getConfigService().getIntValue("business.fluxctrl.receive_queue_per", 50) * 1.0D / 100.0D;
      if (persent > MIN)
      {
        if (Math.random() < (persent - MIN) / (1.0D - MIN))
        {
          logger.warn("connectionfull|" + getSGWChannel().getMaxRecvQueueSize() + "|" + request.getClientAddress());
          XmlMessage ack = createERRAnwser("", 11007, "sgw is busy.");
          ack.setMsgTime(System.currentTimeMillis());
          directAck(ack, body);
          try
          {
            String reqPath = request.getPath().getPath();
            if (reqPath.endsWith(this.standardWSReqPath))
            {
              String ip = "";
              XmlMessage req = null;
              try
              {
                req = parseWSRequest(reqContent);
                if (request.getClientAddress() != null)
                {
                  ip = request.getClientAddress().getAddress().getHostAddress();
                }
              }
              catch (Exception e)
              {
                logger.warn("Unknown http request from " + ip + "." + reqContent);
                return;
              }
              req.setMsgTime(receiveTime);
              req.setIp(ip);

              DCCAsynCallHandler tmp = new DCCAsynCallHandler(null, null, null);
              CallParam callParam = new CallParam(req.getMID(), req.getSCID(), req, req.getUID(), tmp);
              callParam.setRetCode(11007);
              callParam.setResultMsg("sgw is busy.");
              callParam.setResponse(ack);
              Node node = SpringContext.getInstance().getConfigService().getNodeByName(callParam.getUserName());

              if (node == null) {
                logger.warn("AuthFailed: UserName not known request from " + ip + "." + reqContent);
                return;
              }
              callParam.setFromNode(node);
              BusinessRecordCacheManager.getInstance().add(callParam);
            }
          }
          catch (Exception e)
          {
            logger.warn("http request err:", e);
          }
          return;
        }

      }

      String reqPath = request.getPath().getPath();
      if (reqPath.endsWith(this.standardWSReqPath)) {
        doStandardWSCall(receiveTime, request.getClientAddress(), body, reqContent, request, response);
        return;
      }if (reqPath.endsWith(this.standardHttpReqPath)) {
        doStandardHttpCall(receiveTime, body, reqContent, request, response);
        return;
      }
      dealWithOtherReqPath(receiveTime, request, body, reqPath);
      return;
    }
    catch (Exception e) {
      logger.warn("http request err:", e);
      if (body != null) {
        body.println("Server error :" + e.toString());
        body.close();
      }
    }
  }

  private void doStandardWSCall(long reqTime, InetSocketAddress clientAddress, PrintStream body, String reqContent, Request request, Response response)
    throws Exception
  {
    XmlMessage req = null;
    try {
      req = parseWSRequest(reqContent);
    } catch (Exception e) {
      directAck(createERRAnwser("", 11009, e.toString()), body);
      return;
    }
    String serviceCtxID = req.getSCID();
    if ((this.replaceRealm) && (!serviceCtxID.endsWith(this.confSrv.getRealmWithoutAreaCode()))) {
      serviceCtxID = DccAvpHelper.replaceSrvCtxID(serviceCtxID, this.confSrv.getRealmWithoutAreaCode());
      req.setSCID(serviceCtxID);
    }

    BaseWSCallHandler asynHandler = new BaseWSCallHandler(body);
    req.setIp(clientAddress != null ? clientAddress.getAddress().getHostAddress() : "");
    req.setMsgTime(reqTime);    

    CallParam callParam = new CallParam(req.getMID(), req.getSCID(), req, req.getUID(), asynHandler);

    MessageDetail detail = new MessageDetail();
    detail.setBusinessCallId(callParam.getBussinessCallId());
    detail.setSessionId(callParam.getSessionId());
    detail.setServiceContextId(callParam.getServiceContextId());
    detail.setMessageType("WSR");
    detail.setDetails(req.getDATA());
    detail.setMsgTime(new Date(reqTime));
    detail.setOpType("recv");
    detail.setHostId(callParam.getUserName());
    MessageDetailCacheManager.getInstance().add(detail);

    StringBuilder errMsg = new StringBuilder();
    int result = BuzProcessorManager.getInstance().checkBonCallParamForSoap(callParam, errMsg);
    if (result != 0) {
      callParam.setRetCode(result);
      callParam.setResultMsg(errMsg.toString());
      asynHandler.onCallResult(callParam);
      return;
    }

    String cap = DccAvpHelper.getCapability(callParam.getServiceContextId());
    String capAndUsr = cap + callParam.getUserName();
    BlockRequestBean blockBean = (BlockRequestBean)this.confSrv.getRequestBlockMap().get(capAndUsr);
    if (blockBean != null) {
      logger.info("sleep for user with capability " + capAndUsr + " seconds:" + blockBean.getSleepTime());

      long delayMills = System.currentTimeMillis() + blockBean.getSleepTime().intValue() * 1000;
      DelayCallParam delayCallParam = new DelayCallParam(delayMills, callParam);
      this._sessionMap.put(callParam.getSessionId(), delayCallParam);
      logger.info(callParam.getSessionId() + " will forward after " + blockBean.getSleepTime() + " seconds");
      
      return;
    }
    
    // 检查是否走新流程，并处理新流程，若返回[TYPE_ORIG_FLOW]则表示走老流程
    logger.info("start check new flow...");
    logger.info("doProcedure...");
    String retCode = outApiBizProcedure.doProcedure(req, request, response, reqTime);
    if(!BizConsts.TYPE_ORIG_FLOW.equals(retCode)) {
    	logger.info("working on the new flow, and completely.");
    	return;
    }
    logger.info("working on the old flow...");
    
    BuzProcessorManager.getInstance().processRequest(callParam);
    
  }

  private void doStandardHttpCall(long reqTime, PrintStream body, String reqContent, Request request, Response response)
  {
    logger.info("http request of tuxedo call " + reqContent);
    XmlMessage req = null;
    try {
      String[] errcd_and_desc = new String[2];
      req = Comutils.tuxedoXml2SgwQueryReq(reqContent, errcd_and_desc);
      if (req == null) {
        writeHTTPAck("", errcd_and_desc[0], errcd_and_desc[1], "", body);
        return;
      }
      req.setMsgTime(reqTime);
    }
    catch (Exception e) {
      logger.warn("parse tuxedo http post request exception" + e);
      writeHTTPAck("", "11104", "malformed xml:" + e.toString(), "", body);
      return;
    }

    String serviceCtxID = req.getSCID();
    if ((this.replaceRealm) && (!serviceCtxID.endsWith(this.confSrv.getRealmWithoutAreaCode()))) {
      serviceCtxID = DccAvpHelper.replaceSrvCtxID(serviceCtxID, this.confSrv.getRealmWithoutAreaCode());
      req.setSCID(serviceCtxID);
    }

    TuxedoCallHandler asynHandler = new TuxedoCallHandler(body);
    CallParam callParam = new CallParam(req.getMID(), req.getSCID(), req, req.getUID(), asynHandler);

    MessageDetail detail = new MessageDetail();
    detail.setBusinessCallId(callParam.getBussinessCallId());
    detail.setSessionId(callParam.getSessionId());
    detail.setServiceContextId(callParam.getServiceContextId());
    detail.setMessageType("WSR");
    detail.setDetails(req.getDATA());
    detail.setMsgTime(new Date(reqTime));
    detail.setOpType("recv");
    detail.setHostId(callParam.getUserName());
    MessageDetailCacheManager.getInstance().add(detail);

    StringBuilder errMsg = new StringBuilder();
    int result = BuzProcessorManager.getInstance().checkBonCallParamForSoap(callParam, errMsg);
    if (result != 0) {
      callParam.setRetCode(result);
      callParam.setResultMsg(errMsg.toString());
      asynHandler.onCallResult(callParam);
      return;
    }
    
    // 检查是否走新流程，并处理新流程，若返回[TYPE_ORIG_FLOW]则表示走老流程
    logger.info("start check new flow...");
    logger.info("doProcedure...");
    String retCode = outApiBizProcedure.doProcedure(req, request, response, reqTime);
    if(!BizConsts.TYPE_ORIG_FLOW.equals(retCode)) {
    	logger.info("working on the new flow, and completely.");
    	return;
    }
    logger.info("working on the old flow...");

    BuzProcessorManager.getInstance().processRequest(callParam);
  }

  protected void dealWithOtherReqPath(long reqTime, Request request, PrintStream body, String reqPath) throws Exception
  {
    body.println("No service provided at this address path ");
    body.close();
  }

  public void setReplaceRealm(boolean replaceRealm) {
    this.replaceRealm = replaceRealm;
  }

  public void setStandardWSReqPath(String standardWSReqPath) {
    this.standardWSReqPath = standardWSReqPath;
  }

  public void setStandardHttpReqPath(String standardHttpReqPath) {
    this.standardHttpReqPath = standardHttpReqPath;
  }

  @Autowired
  public void setRsaSignService(RSASignService rsaSignService) {
    this.rsaSignService = rsaSignService;
  }

  public Set<String> monitorConfKeys()
  {
    return this.monitorConfKeys;
  }

  public void onConfigUpdated(String updatedKey)
  {
    loadParams();
  }

  private void loadParams()
  {
    this.maxLinkCount = this.confSrv.getIntValue("business.max_conn", 10000);
    logger.info("load param business.max_conn to value " + this.maxLinkCount);
  }

  public int sendRequest(CallParam callParam)
  {
    if (callParam.getToNode() != null) {
      SGWService sgwServiceClient = Comutils.getSGWServiceClient(callParam.getToNode().getServerUrl());

      if (sgwServiceClient == null) {
        callParam.setRetCode(11204);
        callParam.setResultMsg("wsclient is null");
        callParam.getAsynHandler().onCallResult(callParam);
        InprogressMessageQueue.getInstance().removeProgress(callParam.getMessageUniqueID());
        return -1;
      }

      SgwQueryReq sgwreq = new SgwQueryReq();
      sgwreq.setMID(callParam.getSessionId());
      sgwreq.setUID(callParam.getUserName());
      String pwd = callParam.getFromNode().getPassword();
      try {
        sgwreq.setPWD(pwd == null ? null : RSAUtil.getReadableMD5Data(pwd.getBytes("iso-8859-1")));
      } catch (UnsupportedEncodingException e1) {
        e1.printStackTrace();
      }
      sgwreq.setSCID(callParam.getServiceContextId());

      if ((callParam.getRequest() instanceof XmlMessage)) {
        sgwreq.setDATA(((XmlMessage)callParam.getRequest()).getDATA());
      } else if ((callParam.getRequest() instanceof Message))
      {
        AVP sgwRootAVP = DccAvpHelper.createSGWRootAVPFromMessage((Message)callParam.getRequest());
        ProtocalConvertResult result = DataExchangerImpl.getInstance().convertFromCCRRequest(callParam.getServiceContextId(), sgwRootAVP);
        if (result.getRetCode() == 0) {
          sgwreq.setDATA(((Element)result.getResult()).asXML());
        } else {
          callParam.setRetCode(result.getRetCode());
          callParam.setResultMsg(result.getErrMsg());
          callParam.getAsynHandler().onCallResult(callParam);
          InprogressMessageQueue.getInstance().removeProgress(callParam.getMessageUniqueID());
          return -1;
        }
      }

      SgwQueryAck sgwAck = null;
      try {
    	  
	    XmlMessage sendRequest = new XmlMessage(true);
	    sendRequest.setMsgTime(System.currentTimeMillis());
	    sendRequest.setMID(callParam.getSessionId());
	    sendRequest.setSCID(callParam.getServiceContextId());
	    sendRequest.setLength(sgwreq.getDATA().length());
	    sendRequest.setDATA(sgwreq.getDATA());
	    callParam.setSendRequest(sendRequest);
    	   
    	   
        sgwAck = sgwServiceClient.sgwQueryRequest(sgwreq);

        XmlMessage ack = new XmlMessage(false);
        ack.setMsgTime(System.currentTimeMillis());
        ack.setMID(sgwAck.getMID());
        ack.setSIG(sgwAck.getSIG());
        ack.setRC(sgwAck.getRC());
        ack.setDS(sgwAck.getDS());
        ack.setRTT(sgwAck.getRTT());
        ack.setREPTAG(sgwAck.getReptag());
        ack.setDATA(sgwAck.getDATA());
        try
        {
          if ((sgwAck.getDATA() != null) && (!"".equals(sgwAck.getDATA().toString()))) {
            Document doc = Comutils.parseText(sgwAck.getDATA());
            ack.setDataElement(doc.getRootElement());
          }
        }
        catch (Exception localException1)
        {
        }
        callParam.setRecvResponse(ack);

        int callRc = Integer.parseInt(sgwAck.getRC());
        callParam.setRetCode(callRc);
        callParam.setResultMsg(sgwAck.getDS());
      }
      catch (Exception e)
      {
        logger.warn("call ws service err:", e);
        int retCode = LogTool.getErrorCodeForBonCallException(e);
        callParam.setRetCode(retCode);
        callParam.setResultMsg("Call ws service err:" + e.getLocalizedMessage());
      }
      InprogressMessageQueue.getInstance().removeProgress(callParam.getMessageUniqueID());
      callParam.getAsynHandler().onCallResult(callParam);
    }

    return 0;
  }

  public SGWChannel getSGWChannel()
  {
    return this.channel;
  }

  public void stop()
  {
    logger.info("stop SoapChannel: " + this.channel.getChannelHost() + " ...");
    if (this.server != null) {
      try {
        this.server.stop();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
    this.channel.setStatus(0);
    this.confSrv.updateChannelStatus2DB(this.channel.getId(), 0);
  }

  public void addPersist(Node node)
  {
  }

  public void deletePersist(Node node)
  {
  }

  public int getThreadPoolSize()
  {
    return this.executor.getPoolSize();
  }

  public int getThreadPoolQueueSize()
  {
    return this.executor.getQueue().size();
  }

  public int getNodeSendQueueSize(String nodeHostId)
  {
    return 0;
  }

  public int getThreadPoolActiveCount()
  {
    return this.executor.getActiveCount();
  }

  private class DelayCallParam
  {
    private long delayMills;
    private CallParam callParam;

    public DelayCallParam(long delayMills, CallParam callParam)
    {
      this.delayMills = delayMills;
      this.callParam = callParam;
    }
    public long getDelayMills() {
      return this.delayMills;
    }
    public CallParam getCallParam() {
      return this.callParam;
    }
  }

  private class DelayTask extends TimerTask
  {
    private DelayTask() {
    }

    public void run() {
      long now = System.currentTimeMillis() - 500L;
      ArrayList<String> deleteSession = new ArrayList();
      for (Map.Entry entry : SoapChannel.this._sessionMap.entrySet())
      {
        SoapChannel.DelayCallParam delayCallParam = (SoapChannel.DelayCallParam)entry.getValue();
        if (delayCallParam.getDelayMills() < now)
        {
          deleteSession.add((String)entry.getKey());
          try {
            BuzProcessorManager.getInstance().processRequest(delayCallParam.getCallParam());
          } catch (Exception ex) {
            SoapChannel.logger.info(getClass() + " delay exception." + ex);
          }
        }
      }
      for (String session : deleteSession)
      {
        SoapChannel.this._sessionMap.remove(session);
      }
    }
  }
}